from unimport import *

# from pyhive import hive
# hive_conn = hive.Connection(host="172.20.1.229", port=10000, database="dwd", username="gk")

import warnings


def set_float_show(f="{:,.3f}"):
    pd.set_option('display.float_format', f)


def set_warnings(need=True):
    if need:
        warnings.filterwarnings('default')
    else:
        warnings.filterwarnings('ignore')


set_warnings(False)


def set_pd_show(max_rows=500, max_columns=200, max_colwidth=70):
    pd.set_option("display.max_rows", max_rows)
    pd.set_option("display.max_columns", max_columns)
    pd.set_option("display.max_colwidth", max_colwidth)


def mysql_engine(host, port, user, passwd, db=None):
    import pymysql
    from sqlalchemy import create_engine
    try:
        engine = create_engine(f'mysql+pymysql://{user}:{passwd}@{host}:{port}/{db}')
        return engine
    except Exception as e:
        print(f"An error occurred: {e}")
    # conn_lm = mysql_engine('172.20.1.186', 3306, 'longxiaolei', 'WWsCQ9TXG+BJBHlK', 'rule')  #拉美
    # conn_fz= mysql_engine('110.238.79.30', 4000, 'longxiaolei', 'nOc2v+ariqTq+01L', 'at_system') # 非洲
    # conn_yz= mysql_engine('8.215.28.149', 4000, 'leo', 'ojSLOkv0DC2ywkxu', 'af_system') # 亚洲


def week_start_day(date_obj):
    """
    获取该日期对象，对应的每周第一天
    :param date_obj:
    :return:
    """
    weekday = date_obj.weekday()
    start_of_week = date_obj - timedelta(days=weekday)
    return start_of_week.strftime('%Y-%m-%d')


def format_date2week(date_obj):
    """
    将日期对象，转化为周
    :param date_obj:
    :return:
    """
    weekday = date_obj.weekday()
    start_of_week = date_obj - timedelta(days=weekday)
    end_of_week = start_of_week + timedelta(days=6)
    monday = start_of_week.strftime('%m.%d')
    year = start_of_week.year
    sunday = end_of_week.strftime('%m.%d')
    return f'{year}.{monday}~{sunday}'


def month_window(window_size: int, start_month='2023-01', end_month=None):
    """
    根据窗口大小和起始时间，计算滑动月份的滑动时间窗
    :param window_size: 窗口大小
    :param start_month: 开始月份
    :param end_month: 结束的月份，默认为当前月份
    :return:
    """
    windows = []
    date_time = datetime.strptime(start_month, "%Y-%m")
    end_month = end_month if end_month else datetime.now().strftime("%Y-%m")
    end_date = datetime.strptime(end_month, "%Y-%m")
    while date_time <= end_date:
        start_time = (date_time - relativedelta(months=1))
        end_time = (date_time - relativedelta(months=window_size))
        windows.append((date_time.strftime('%Y-%m'), start_time.strftime('%Y-%m'), end_time.strftime('%Y-%m')))
        date_time = date_time + relativedelta(months=1)
    return windows


def continue_months(start_month='2023-01', end_month=None):
    date_time = datetime.strptime(start_month, "%Y-%m")
    end_month = end_month if end_month else datetime.now().strftime("%Y-%m")
    end_date = datetime.strptime(end_month, "%Y-%m")
    months = []
    while date_time <= end_date:
        months.append(date_time.strftime('%Y-%m'))
        date_time = date_time + relativedelta(months=1)
    return months




def is_decimal(string):
    """
    判断字符串是否是数字格式
    :param string:
    :return:
    """
    pattern = r'^-?\d+\.\d+$'
    return bool(re.match(pattern, string))


def get_decimal_places(number):
    """
    获取小数部分的位数
    :param number:
    :return:
    """
    # 设置高精度
    getcontext().prec = 50
    decimal_number = Decimal(str(number))

    # 去掉整数部分
    decimal_part = decimal_number - decimal_number.to_integral_value()

    # 获取小数部分的位数
    if decimal_part == Decimal(0):
        return 0
    else:
        return abs(decimal_part.as_tuple().exponent)


def ngrams_code(tokens, n, join_char=" "):
    """
    对输入的数组进行ngram编码
    :param tokens: 需要进行ngram编码的数组
    :param n:
    :param join_char: 连接字符，默认空格符
    :return:
    """
    return [join_char.join(tokens[i:i + n]) for i in range(len(tokens) - n + 1)]


def parallel_process(task_function, task_list, process_num=10):
    """
    并发跑批函数
    :param task_function:
    :param task_list:
    :param process_num:
    :return:
    """

    def _task(task_function, queue, task_params):
        try:
            task_function(*task_params)
        finally:
            queue.put(True)

    from multiprocessing import Queue, Process
    tq = Queue()
    for i in range(process_num):
        tq.put(True)
    for task_params in tqdm(task_list):
        tq.get()
        Process(target=_task, args=(task_function, tq, task_params)).start()
    for i in range(process_num):
        tq.get()
    tq.close()
    gc.collect()
